package com.bwie.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.io.file.FileNameUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.bwie.domain.TbDict;
import com.bwie.domain.TbLog;
import com.bwie.domain.TbProd;
import com.bwie.es.EsProd;
import com.bwie.mapper.TbDictMapper;
import com.bwie.mapper.TbLogMapper;
import com.bwie.mapper.TbProdMapper;
import com.bwie.service.TbProdService;
import com.bwie.state.DictEnum;
import com.bwie.state.ProdEnum;
import com.bwie.state.UserRoleEnum;
import com.bwie.utils.MobileMessageUtil;
import com.bwie.utils.R;
import com.bwie.vo.MessageVo;
import com.bwie.vo.PageInfoVo;
import com.github.tobato.fastdfs.domain.StorePath;
import com.github.tobato.fastdfs.service.FastFileStorageClient;
import io.jsonwebtoken.Claims;
import io.jsonwebtoken.Jwts;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import javax.annotation.PostConstruct;
import javax.servlet.http.HttpServletRequest;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;

/**
 * @author 军哥
 * @version 1.0
 * @description: TODO
 * @date 2024/11/14 16:31
 */

@Service
public class TbProdServiceImpl implements TbProdService {

    @Autowired
    TbProdMapper tbProdMapper;

    @Autowired
    TbDictMapper tbDictMapper;

    @Autowired
    TbLogMapper tbLogMapper;

    @Autowired
    FastFileStorageClient fastFileStorageClient;

    @Autowired
    HttpServletRequest request;

    @Autowired
    KafkaTemplate kafkaTemplate;

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Autowired
    ElasticsearchRestTemplate elasticsearchRestTemplate;

    /**
     * 项目启动时，加载字典数据
     * @PostConstruct 这个注解，就是项目启动时，运行，并且只运行一次
     */
    @PostConstruct
    public void loadDictList() {
        System.out.println("加载字典数据~~~~~~~~~~~~~~~~~~");
        // 获取key和数据
        String key = "key-dictList";
        List<TbDict> tbDicts = tbDictMapper.selectList(null);

        // 删除缓存
        stringRedisTemplate.delete(key);

        // 添加缓存
        stringRedisTemplate.opsForValue().set(key, JSON.toJSONString(tbDicts));
    }

    @Override
    public R getProdList(PageInfoVo pageInfoVo) {
        // 获取token
        String token = request.getHeader("token");
        if(token == null) {
            return R.ERROR(500, "请先登录");
        }
        Claims body = Jwts.parser().setSigningKey("123456").parseClaimsJws(token).getBody();
        Integer userId = body.get("userId", Integer.class);
        Integer userRole = body.get("userRole", Integer.class);


        //--1 构造分页
        Page<TbProd> page = new Page<>(pageInfoVo.getPageNum(), pageInfoVo.getPageSize());

        // 添加搜索条件：模糊搜索 null,''
        QueryWrapper<TbProd> wrapper = new QueryWrapper<>();
        if(!StringUtils.isEmpty(pageInfoVo.getKeyWord())) {
            wrapper.lambda().like(TbProd::getProdName, pageInfoVo.getKeyWord());
        }

        // 根据用户ID获取商品列表：商户只能看到自己的商品
        // 管理员要能看到所有商品，普通用户用户，只能看到自己的商品
        if(userRole.equals(UserRoleEnum.SHOP.getCode())) {
            wrapper.lambda().eq(TbProd::getUserId, userId);
        }

        // 精确查询
        if(pageInfoVo.getTypeId() != null) {
            wrapper.lambda().eq(TbProd::getProdStatus, pageInfoVo.getTypeId());
        }

        // 时间区间查询
        if(!StringUtils.isEmpty(pageInfoVo.getBeginTime())) {
            wrapper.lambda().gt(TbProd::getCreateTime, pageInfoVo.getBeginTime());
        }
        if(!StringUtils.isEmpty(pageInfoVo.getEndTime())) {
            wrapper.lambda().lt(TbProd::getCreateTime, pageInfoVo.getEndTime());
        }

        // 根据商品录入时间排序
        wrapper.lambda().orderByDesc(TbProd::getCreateTime);

        //--2 获取列表
        Page<TbProd> tbProdPage = tbProdMapper.selectPage(page, wrapper);
        return R.OK(tbProdPage);
    }

    @Override
    public R getProdStatusList() {

        String key = "key-dictList-" + DictEnum.PROD_STATUS.getCode();
        Boolean hasKey = stringRedisTemplate.hasKey(key);

        if(hasKey) {
            // 缓存中有数据，从缓存中读取
            System.out.println("缓存中有数据，从缓存中读取");
            String json = stringRedisTemplate.opsForValue().get(key);
            List<TbDict> tbDicts = JSON.parseArray(json, TbDict.class);
            return R.OK(tbDicts);
        }
        else {
            // 缓存中没有数据，从数据库中读取
            System.out.println("缓存中没有数据，从数据库中读取");
            List<TbDict> tbDicts = tbDictMapper.selectList(
                new QueryWrapper<TbDict>().lambda().eq(TbDict::getDictType, DictEnum.PROD_STATUS.getCode())
            );

            // 把数据存入缓存
            stringRedisTemplate.opsForValue().set(key, JSON.toJSONString(tbDicts));

            // 返回数据
            return R.OK(tbDicts);
        }
    }

    @Override
    public R uploadFile(MultipartFile file) {
        try {
            String extName = FileNameUtil.extName(file.getOriginalFilename());
            StorePath storePath = fastFileStorageClient.uploadFile(file.getInputStream(), file.getSize(), extName, null);
            String url = "http://192.168.80.192:8888/" + storePath.getFullPath();
            return R.OK(url);
        } catch (IOException e) {
            e.printStackTrace();
        }

        return R.ERROR(500, "上传失败");
    }

    @Override
    public R addProd(TbProd tbProd) {
        //-- 检查是否唯一
        TbProd prod = tbProdMapper.selectOne(
                new QueryWrapper<TbProd>().lambda().eq(TbProd::getProdName, tbProd.getProdName())
        );
        if(prod != null) {
            return R.ERROR(500, "商品名称已存在");
        }

        //-- 添加商户信息
        String token = request.getHeader("token");
        Claims body = Jwts.parser().setSigningKey("123456").parseClaimsJws(token).getBody();
        Integer userId = body.get("userId", Integer.class);
        tbProd.setUserId(userId);

        //-- 添加商品到数据库中
        tbProd.setProdStatus(0);
        tbProd.setCreateTime(new Date());
        tbProdMapper.insert(tbProd);

        //-- 把商品信息，添加到ES中
        // save 方法 即是添加，又是修改，根据ID来决定
        EsProd esProd = BeanUtil.toBean(tbProd, EsProd.class);
        elasticsearchRestTemplate.save(esProd);

        return R.OK();
    }

    @Override
    public R updateProd(TbProd tbProd) {
        //-- 检查是否存在
        TbProd prod = tbProdMapper.selectById(tbProd.getProdId());
        if(prod == null) {
            return R.ERROR(500, "商品名称不存在");
        }

        //-- 修改数据库中的商品
        tbProdMapper.updateById(tbProd);

        //-- 修改ES中的商品
        tbProd = tbProdMapper.selectById(tbProd.getProdId());
        EsProd esProd = BeanUtil.toBean(tbProd, EsProd.class);
        elasticsearchRestTemplate.save(esProd);

        return R.OK();
    }

    @Override
    public R deleteProd(Integer prodId) {
        //-- 检查是否存在
        TbProd prod = tbProdMapper.selectById(prodId);
        if(prod == null) {
            return R.ERROR(500, "商品名称不存在");
        }

        // 从数据库中删除商品
        tbProdMapper.deleteById(prodId);

        // 从ES中删除商品
        elasticsearchRestTemplate.delete(""+prodId, EsProd.class);

        return R.OK();
    }

    @Override
    public R sendKafkaMessage(String message) {

        // 第一个参数：主题topic
        // 第二个参数： 发送的消息（最好发送的是字符串）
        kafkaTemplate.send("topic-test", message);

        return R.OK();
    }

    @Override
    public R uploadProd(Integer prodId) {
        //--1 获取商品信息
        TbProd tbProd = tbProdMapper.selectById(prodId);
        if(tbProd == null) {
            return R.ERROR(500, "商品不存在");
        }

        //--2 获取用户ID，判断商品是否属于这个商家
        String token = request.getHeader("token");
        if(token == null) {
            return R.ERROR(500, "请先登录");
        }
        Claims body = Jwts.parser().setSigningKey("123456").parseClaimsJws(token).getBody();
        Integer userId = body.get("userId", Integer.class);
        if(!userId.equals(tbProd.getUserId())) {
            return R.ERROR(500, "商品不属于当前用户");
        }

        //--3 商品上架
        tbProd.setProdStatus(ProdEnum.UP.getCode());
        tbProdMapper.updateById(tbProd);

        //--4 发送消息，记录商家日志
        TbLog tbLog = new TbLog();
        tbLog.setLogTitle("商品上架");
        tbLog.setLogContent("商品信息：" + JSON.toJSONString(tbProd));
        String ip = request.getRemoteAddr().toString();
        tbLog.setLogIp(ip);
        tbLog.setUserId(userId);
        String userName = body.get("userName", String.class);
        tbLog.setUserName(userName);
        tbLog.setCreateTime(new Date());

        kafkaTemplate.send("topic-prod", JSON.toJSONString(tbLog));

        return R.OK();
    }

    @Override
    public R getProdListFromEs(PageInfoVo pageInfoVo) {
        //--1 构造分页参数
        Integer pageNum = 0;
        if(pageInfoVo.getPageNum() > 0) {
            pageNum = pageInfoVo.getPageNum() - 1;
        }
        PageRequest pageRequest = PageRequest.of(pageNum, pageInfoVo.getPageSize());

        //--2 构造查询条件
        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();

        // 根据商品名称搜索
        // matchQuery 分词搜索 学习和游泳，学习，游泳
        if(!StringUtils.isEmpty(pageInfoVo.getKeyWord())) {
            boolQueryBuilder.must(QueryBuilders.matchQuery("prodName", pageInfoVo.getKeyWord()));
        }

        // 数值区间搜索
        // rangeQuery 区间查询 g greater 大于， l little 小于, e equal 等于
        if(pageInfoVo.getMinValue() != null && pageInfoVo.getMaxValue() != null) {
            boolQueryBuilder.must(
                    QueryBuilders.rangeQuery("prodPrice")
                            .gte(pageInfoVo.getMinValue())
                            .lte(pageInfoVo.getMaxValue())
            );
        }

        // 时间区间查询
        if(!StringUtils.isEmpty(pageInfoVo.getBeginTime()) && !StringUtils.isEmpty(pageInfoVo.getEndTime())) {
            boolQueryBuilder.must(
                    QueryBuilders.rangeQuery("createTime")
                            .format("yyyy-MM-dd HH:mm:ss")
                            .timeZone("GMT+8")
                            .gte(pageInfoVo.getBeginTime())
                            .lte(pageInfoVo.getEndTime())
            );
        }

        // 添加高亮
        HighlightBuilder highlightBuilder = new HighlightBuilder();
        highlightBuilder.field("prodName").preTags("<font color='red'>").postTags("</font>");

        //--3 执行查询
        NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
        nativeSearchQueryBuilder.withQuery(boolQueryBuilder)
                .withPageable(pageRequest)
                .withHighlightBuilder(highlightBuilder);

        SearchHits<EsProd> searchHits = elasticsearchRestTemplate.search(nativeSearchQueryBuilder.build(), EsProd.class);

        //--4 读取查询结果，返回数据
        long total = searchHits.getTotalHits();

        ArrayList<EsProd> records = new ArrayList<>();
        List<SearchHit<EsProd>> hits = searchHits.getSearchHits();
        for (SearchHit<EsProd> hit : hits) {
            EsProd esProd = hit.getContent();

            // 获取高亮字段
            List<String> strings = hit.getHighlightField("prodName");
            if(strings != null && strings.size() > 0) {
                esProd.setProdName(strings.get(0));
            }

            records.add(esProd);
        }

        HashMap<String, Object> map = new HashMap<>();
        map.put("total", total);
        map.put("records", records);

        return R.OK(map);
    }

    @KafkaListener(topics = "topic-prod")
    public void recvProdTopicMessage(String message, Acknowledgment acknowledgment) {
        System.out.println("接收到消息：" + message);
        //-- 接收消息
        TbLog tbLog = JSON.parseObject(message, TbLog.class);

        // 处理消息
        tbLogMapper.insert(tbLog);

        // 手动确认消息，并删除
        acknowledgment.acknowledge();
    }

    /**
     * 接收审核不通过的消息
     * @param message
     * @param acknowledgment
     */
    @KafkaListener(topics = "topic-check")
    public void recvCheckMessage(String message, Acknowledgment acknowledgment) {
        System.out.println("接收到审核不通过的消息：" + message);

        MessageVo messageVo = JSON.parseObject(message, MessageVo.class);
        System.out.println("消息ID：" + messageVo.getMsgId());

        String msgBody = messageVo.getMsgBody();
        JSONObject jsonBody = JSON.parseObject(msgBody);

        MobileMessageUtil.sendMobileMessage(jsonBody.getString("mobile"), jsonBody.getString("content"));
        acknowledgment.acknowledge();
        System.out.println("消息已确认，短信已发送");
    }



    @KafkaListener(topics = "topic-test")
    public void recvTestTopicMessage(String message, Acknowledgment acknowledgment) {
        // 接收消息
        System.out.println("接收到消息：" + message);
        // 手动确认消息，并删除
        acknowledgment.acknowledge();
    }
}
